#include "Ht_Hq_ThreadPool.h"

#include <assert.h>


CHt_Hq_ThreadPool::CHt_Hq_ThreadPool():
m_iMaxThreadNum(50),
m_iAvailLow(5),
m_iAvailHigh(20),
m_iUseThreadNum(10),
m_iInitThreadNum(10)
{
	m_vAllThreads.clear();
	m_vBusyThreads.clear();
	m_vIdleThreads.clear();

	for (int i = 0; i < m_iInitThreadNum; i++)
	{
		CWorkThread *pWorkThread = new CWorkThread();
		pWorkThread->SetThreadPool(this);
		AppendToIdleList(pWorkThread);
		pWorkThread->Start();
	}
}


CHt_Hq_ThreadPool::CHt_Hq_ThreadPool(int iNum)
{
	assert(iNum > 0 && iNum <= 30);

	m_iMaxThreadNum = 30;
	m_iAvailLow= iNum - 10 > 0 ? iNum - 10 : 3;
	m_iInitThreadNum = m_iUseThreadNum = iNum;
	m_iAvailHigh = iNum + 10;

	m_vBusyThreads.clear();
	m_vIdleThreads.clear();
	m_vAllThreads.clear();

	for (int i = 0; i < m_iInitThreadNum; i++)
	{
		CWorkThread *pThread = new CWorkThread();
		pThread->SetThreadPool(this);
		AppendToIdleList(pThread);
		pThread->Start();
	}
}

CHt_Hq_ThreadPool::~CHt_Hq_ThreadPool()
{
	TerminateAll();
}

void CHt_Hq_ThreadPool::TerminateAll()
{
	for (size_t i = 0; i < m_vAllThreads.size(); i++)
	{
		CWorkThread *pThread = m_vAllThreads.at(i);

		pThread->Terminate();
	}
}

void CHt_Hq_ThreadPool::Run(CJob *pJob, void * pJobData)
{
	//assert(NULL == pJob);

	if (m_iMaxThreadNum <= this->GetBusyThreadsNum())
		m_valCond.Wait();

	if (m_vIdleThreads.size() < (size_t)m_iAvailLow)
	{
		if (GetAllThreadsNum() + GetInitThreadNum() - (int)m_vIdleThreads.size() < m_iMaxThreadNum)
			CreatIdleThread(m_iInitThreadNum - m_vIdleThreads.size());
		else
			CreatIdleThread(GetMaxThreadNum() - GetAllThreadsNum());
	}

	CWorkThread *pWorkThread = GetIdleThread();
	if (NULL != pWorkThread)
	{
		pWorkThread->m_workMutex.Lock();
		MoveToBusyThreadList(pWorkThread);
		pWorkThread->SetThreadPool(this);
		pJob->SetWorkThread(pWorkThread);
		pWorkThread->AddNewJob(pJob, pJobData);
		pWorkThread->m_workMutex.UnLock();
	}
}

CWorkThread * CHt_Hq_ThreadPool::GetIdleThread()
{
	while (0 == m_vIdleThreads.size())
	{
		m_idleCond.Wait();
	}

	m_idleMutex.Lock();
	if (m_vIdleThreads.size() > 0)
	{
		CWorkThread * pThread = m_vIdleThreads.front();
		m_idleMutex.UnLock();
		return pThread;
	}
	m_idleMutex.UnLock();

	return NULL;
}

void CHt_Hq_ThreadPool::MoveToIdleThreadList(CWorkThread* pThread)
{
	m_idleMutex.Lock();
	m_vIdleThreads.push_back(pThread);
	m_iUseThreadNum++;
	m_idleMutex.UnLock();

	m_busyMutex.Lock();
	vector<CWorkThread *>::iterator it;
	it = find(m_vBusyThreads.begin(), m_vBusyThreads.end(), pThread);
	if (it != m_vBusyThreads.end())
		m_vBusyThreads.erase(it);
	m_busyMutex.UnLock();

	m_idleCond.Signal();
	m_valCond.Signal();
}

void CHt_Hq_ThreadPool::MoveToBusyThreadList(CWorkThread* pThread)
{
	m_busyMutex.Lock();
	m_vBusyThreads.push_back(pThread);
	m_iUseThreadNum --;
	m_busyMutex.UnLock();

	m_idleMutex.Lock();
	vector<CWorkThread *>::iterator it;
	it = find(m_vIdleThreads.begin(), m_vIdleThreads.end(), pThread);
	if (it != m_vIdleThreads.end())
		m_vIdleThreads.erase(it);
	m_idleMutex.UnLock();
}

void CHt_Hq_ThreadPool::DeleteIdleThread(int iNum)
{
	m_idleMutex.Lock();
	for (int i = 0; i < iNum; i++)
	{
		CWorkThread *pWorkThread = NULL;
		if (m_vIdleThreads.size() > 0)
		{
			pWorkThread = m_vIdleThreads.front();
		}

		vector<CWorkThread *>::iterator it;

		it = find(m_vIdleThreads.begin(), m_vIdleThreads.end(), pWorkThread);
		if (it != m_vIdleThreads.end())
		{
			m_vIdleThreads.erase(it);
			m_iUseThreadNum --;
		}
	}
	m_idleMutex.UnLock();
}

void CHt_Hq_ThreadPool::CreatIdleThread(int iNum)
{
	for (int i = 0; i < iNum; i++)
	{
		CWorkThread * pThread = new CWorkThread();

		pThread->SetThreadPool(this);
		AppendToIdleList(pThread);

		m_valMutex.Lock();
		m_iUseThreadNum++;
		m_valMutex.UnLock();
		pThread->Start();
	}
}

void CHt_Hq_ThreadPool::AppendToIdleList(CWorkThread * pThread)
{
	m_idleMutex.Lock();
	m_vIdleThreads.push_back(pThread);
	m_vAllThreads.push_back(pThread);
	m_idleMutex.UnLock();
}

//CWorkThread

CWorkThread::CWorkThread()
{
	m_pJob = NULL;
	m_pJobData = NULL;
	m_pThreadPool = NULL;
}

CWorkThread::~CWorkThread()
{
	if (NULL != m_pJob)
	{
		delete m_pJob;
		m_pJob = NULL;
	}

	if (NULL != m_pThreadPool)
	{
		delete m_pThreadPool;
		m_pThreadPool = NULL;
	}
}

void CWorkThread::Run()
{
	cout << "workThread run..." << endl;
	SetThreadStatus(EM_THREAD_RUN);

	while (true)
	{
		while ((NULL == m_pJob) && !GetExitFlag())
		{
			//cout << "thread : " << GetThreadID() << "wait for job." << endl;
			m_jobAddCond.Wait();
			Sleep(100);
		}

		if (NULL == m_pJob)
		{
			if (GetExitFlag())
				break;
			else
				continue;
		}

		m_workMutex.Lock();
		m_pJob->Execute(m_pJobData);
		m_pJob->SetWorkThread(NULL);
		m_pThreadPool->MoveToIdleThreadList(this);
		SetThreadStatus(EM_THREAD_FREE);
		if (m_pThreadPool->m_vIdleThreads.size() > (size_t)m_pThreadPool->GetAvailHigh())
			m_pThreadPool->DeleteIdleThread(m_pThreadPool->m_vIdleThreads.size() - m_pThreadPool->GetAvailHigh());

		m_pJob = NULL;
		m_pThreadPool = NULL;
		m_workMutex.UnLock();
	}
}

void CWorkThread::Terminate()
{
	m_workMutex.Lock();
	SetExitFlag(true);
	m_pJob = NULL;
	m_pJobData = NULL;
	m_jobAddCond.Signal();
	m_workMutex.UnLock();

	::WaitForSingleObject(GetThreadHandle(), INFINITE);
	::CloseHandle(GetThreadHandle());
}

void CWorkThread::AddNewJob(CJob *pJob, void *pJobData)
{
	m_valMutex.Lock();

	m_pJob = pJob;
	m_pJobData = pJobData;
	m_pJob->SetWorkThread(this);

	m_valMutex.UnLock();

	m_jobAddCond.Signal();
}